package org.robby.hbase.cdr.filter;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;

import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * 自定义过滤器
 * 
 * 执行命令：export HBASE_CLASSPATH="/home/conkeyn/jar/custom_filter.jar"
 * 
 * @author conkeyn
 * 
 */
public class CustomFilter implements Filter {

	byte[] value;

	// 必须实现两个构造函数。
	// 一个无参函数
	CustomFilter() {
		super();
	}

	// 一个参带数的构造函数
	CustomFilter(String value) {
		this.value = value.getBytes();
	}

	@Override
	// 传入参数时使用的
	public void readFields(DataInput in) throws IOException {
		this.value = Bytes.readByteArray(in);
	}

	@Override
	public void write(DataOutput out) throws IOException {
		Bytes.writeByteArray(out, value);
		;
	}

	@Override
	public boolean filterAllRemaining() {

		return false;
	}

	@Override
	// SKIP表示不需要将这列返回给客户端，需要跳过这一列
	// INCLUDE表示要将这列返回给客户端
	public ReturnCode filterKeyValue(KeyValue arg0) {

		return ReturnCode.INCLUDE;
	}

	@Override
	public boolean filterRow() {

		return false;
	}

	@Override
	public void filterRow(List<KeyValue> arg0) {

	}

	@Override
	// 返回值为true表示要过滤掉
	// 返回值为false表示不过滤，也就是要这一行记录
	public boolean filterRowKey(byte[] arg0, int arg1, int arg2) {
		for (int i = 0; i < value.length; i++) {
			if (arg0[i] != value[i]) {
				return true;
			}
		}
		return false;
	}

	@Override
	public KeyValue getNextKeyHint(KeyValue arg0) {

		return arg0;
	}

	@Override
	public boolean hasFilterRow() {

		return false;
	}

	@Override
	// 在过滤每行数据之前都会调用一次，用于初始化和重置数据工作
	public void reset() {

	}

	@Override
	// 必需将参数arg0返回或者做转换后再返回。
	public KeyValue transform(KeyValue arg0) {

		return arg0;
	}

}
